#include "improved_pubsub_framework.hpp"
#include "debug_utils.h"
#include <string>
#include <vector>
#include <thread>
#include <chrono>
#include <signal.h>
#include <sys/wait.h>

using namespace improved_pubsub;

// 全局变量用于信号处理
std::unique_ptr<TopicRegistry> g_registry;
std::atomic<bool> g_shutdown{false};

void signal_handler(int signal) {
    DEMO_WARN("signal_handler: 收到信号 {}, 正在关闭...", signal);
    g_shutdown = true;
    if (g_registry) {
        g_registry->stop();
    }
}

// 注册中心进程
void run_registry() {
    DEMO_INFO("run_registry: 启动注册中心进程");
    
    signal(SIGINT, signal_handler);
    signal(SIGTERM, signal_handler);
    DEMO_DEBUG("run_registry: 注册信号处理器");
    
    Config config;
    g_registry = std::make_unique<TopicRegistry>(config);
    DEMO_INFO("run_registry: 创建TopicRegistry实例, socket_path={}", config.registry_socket_path);
    
    if (!g_registry->start()) {
        DEMO_CRITICAL("run_registry: 启动注册中心失败");
        return;
    }
    
    DEMO_INFO("run_registry: 注册中心启动成功, 开始服务循环");
    
    // 保持服务运行
    while (!g_shutdown) {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    
    DEMO_INFO("run_registry: 收到关闭信号, 正在关闭注册中心");
}

// 发布者进程
void run_publisher() {
    DEMO_INFO("run_publisher: 启动发布者进程");
    
    // 等待注册中心启动
    DEMO_DEBUG("run_publisher: 等待注册中心启动...");
    std::this_thread::sleep_for(std::chrono::seconds(1));
    
    Publisher publisher;
    DEMO_DEBUG("run_publisher: 创建Publisher实例");
    
    // 连接到多个topic
    std::vector<std::string> topics = {"news", "weather", "stock"};
    DEMO_INFO("run_publisher: 准备连接到 {} 个topic", topics.size());
    
    for (const auto& topic : topics) {
        DEMO_DEBUG("run_publisher: 尝试连接topic '{}'", topic);
        if (publisher.connect_to_topic(topic)) {
            DEMO_INFO("run_publisher: 成功连接topic '{}'", topic);
        } else {
            DEMO_ERROR("run_publisher: 连接topic '{}'失败", topic);
            return;
        }
    }
    
    // 发布消息
    DEMO_INFO("run_publisher: 开始发布消息循环");
    int message_id = 0;
    while (!g_shutdown && message_id < 15) {
        DEMO_DEBUG("run_publisher: 发布消息轮次 {}", message_id);
        for (const auto& topic : topics) {
            std::string message_text = "改进版消息来自" + topic + " #" + std::to_string(message_id);
            std::vector<uint8_t> message_data(message_text.begin(), message_text.end());
            
            if (publisher.publish(topic, message_data)) {
                DEMO_TRACE("run_publisher: 发布成功, topic='{}', size={}", topic, message_data.size());
            } else {
                DEMO_ERROR("run_publisher: 发布失败, topic='{}', message='{}'", topic, message_text);
            }
            
            std::this_thread::sleep_for(std::chrono::milliseconds(500));
        }
        message_id++;
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
    
    DEMO_INFO("run_publisher: 发布者进程退出");
}

// 订阅者进程
void run_subscriber(const std::string& subscriber_name, const std::vector<std::string>& topics) {
    DEMO_INFO("run_subscriber: 启动订阅者进程 '{}'", subscriber_name);
    
    // 等待注册中心启动
    DEMO_DEBUG("run_subscriber: 等待注册中心启动...");
    std::this_thread::sleep_for(std::chrono::seconds(1));
    
    Subscriber subscriber;
    subscriber.start();
    DEMO_DEBUG("run_subscriber: 启动Subscriber服务");
    
    // 订阅多个topic
    DEMO_INFO("run_subscriber: '{}' 准备订阅 {} 个topic", subscriber_name, topics.size());
    for (const auto& topic : topics) {
        DEMO_DEBUG("run_subscriber: '{}' 尝试订阅topic '{}'", subscriber_name, topic);
        bool success = subscriber.subscribe(topic, [subscriber_name, topic](const Message& msg) {
            std::string message_text(msg.data.begin(), msg.data.end());
            DEMO_TRACE("run_subscriber: '{}' 处理消息, topic='{}', size={}, content='{}'", 
                       subscriber_name, topic, msg.data.size(), message_text);
        });
        
        if (success) {
            DEMO_INFO("run_subscriber: '{}' 成功订阅topic '{}'", subscriber_name, topic);
        } else {
            DEMO_ERROR("run_subscriber: '{}' 订阅topic '{}'失败", subscriber_name, topic);
            return;
        }
    }
    
    DEMO_INFO("run_subscriber: '{}' 开始监听消息循环", subscriber_name);
    
    // 保持订阅者运行
    while (!g_shutdown) {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    
    DEMO_INFO("run_subscriber: '{}' 停止订阅者服务", subscriber_name);
    subscriber.stop();
}

// 演示多进程通信
void demonstrate_improved_multiprocess_pubsub() {
    DEMO_INFO("demonstrate_improved_multiprocess_pubsub: 开始多进程演示");
    std::vector<pid_t> child_pids;
    
    // 启动注册中心
    DEMO_DEBUG("demonstrate_improved_multiprocess_pubsub: fork注册中心进程");
    pid_t registry_pid = fork();
    if (registry_pid == 0) {
        run_registry();
        exit(0);
    } else if (registry_pid > 0) {
        child_pids.push_back(registry_pid);
        DEMO_INFO("demonstrate_improved_multiprocess_pubsub: 注册中心进程启动, pid={}", registry_pid);
    }
    
    // 等待注册中心启动
    DEMO_DEBUG("demonstrate_improved_multiprocess_pubsub: 等待注册中心启动...");
    std::this_thread::sleep_for(std::chrono::seconds(2));
    
    // 启动订阅者1 (订阅news和weather)
    pid_t sub1_pid = fork();
    if (sub1_pid == 0) {
        run_subscriber("订阅者1", {"news", "weather"});
        exit(0);
    } else if (sub1_pid > 0) {
        child_pids.push_back(sub1_pid);
    }
    
    // // 启动订阅者2 (订阅stock和weather)
    // pid_t sub2_pid = fork();
    // if (sub2_pid == 0) {
    //     run_subscriber("订阅者2", {"stock", "weather"});
    //     exit(0);
    // } else if (sub2_pid > 0) {
    //     child_pids.push_back(sub2_pid);
    // }
    
    // // 启动订阅者3 (订阅所有topic)
    // pid_t sub3_pid = fork();
    // if (sub3_pid == 0) {
    //     run_subscriber("订阅者3", {"news", "weather", "stock"});
    //     exit(0);
    // } else if (sub3_pid > 0) {
    //     child_pids.push_back(sub3_pid);
    // }
    
    // 等待订阅者启动
    std::this_thread::sleep_for(std::chrono::seconds(2));
    
    // 启动发布者
    pid_t pub_pid = fork();
    if (pub_pid == 0) {
        run_publisher();
        exit(0);
    } else if (pub_pid > 0) {
        child_pids.push_back(pub_pid);
    }
    
    // 父进程等待用户输入或发布者完成
    DEMO_INFO("demonstrate_improved_multiprocess_pubsub: 主进程监控中，等待自动结束");
    
    // 设置非阻塞输入检查
    auto start_time = std::chrono::steady_clock::now();
    auto max_duration = std::chrono::seconds(100); // 最大运行30秒
    
    while (true) {
        // 检查是否超时
        auto elapsed = std::chrono::steady_clock::now() - start_time;
        if (elapsed >= max_duration) {
            DEMO_INFO("demonstrate_improved_multiprocess_pubsub: 演示时间到，正在结束");
            break;
        }
        
        // 检查发布者是否结束
        int status;
        pid_t result = waitpid(pub_pid, &status, WNOHANG);
        if (result == pub_pid) {
            DEMO_INFO("demonstrate_improved_multiprocess_pubsub: 发布者已完成，等待3秒后结束演示");
            std::this_thread::sleep_for(std::chrono::seconds(3));
            break;
        }
        
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    
    // 结束所有子进程
    DEMO_INFO("demonstrate_improved_multiprocess_pubsub: 正在结束所有进程");
    g_shutdown = true;
    
    for (pid_t pid : child_pids) {
        kill(pid, SIGTERM);
    }
    
    // 等待所有子进程结束
    for (pid_t pid : child_pids) {
        int status;
        waitpid(pid, &status, 0);
        DEMO_INFO("demonstrate_improved_multiprocess_pubsub: 进程 {} 已结束", pid);
    }
    
    DEMO_INFO("demonstrate_improved_multiprocess_pubsub: 演示完成！");
    DEMO_INFO("改进总结: 解决了EventFD竞争问题（每个订阅者独立EventFD）");
    DEMO_INFO("改进总结: 使用msgq模块而非自定义共享内存");
    DEMO_INFO("改进总结: 更清晰的架构分离");
    DEMO_INFO("改进总结: 更好的线程安全性");
}

// 性能测试
void performance_test() {
    DEMO_INFO("performance_test: 开始改进版性能测试");
    
    Publisher publisher;
    if (!publisher.connect_to_topic("perf_test")) {
        DEMO_ERROR("performance_test: 连接topic失败");
        return;
    }
    
    const int message_count = 100000;
    const size_t message_size = 1024*10; // 1KB消息
    
    std::vector<uint8_t> test_data(message_size, 0x42);
    
    auto start_time = std::chrono::high_resolution_clock::now();
    
    for (int i = 0; i < message_count; i++) {
        publisher.publish("perf_test", test_data);
    }
    
    auto end_time = std::chrono::high_resolution_clock::now();
    auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end_time - start_time);
    
    double throughput = (double)message_count / duration.count() * 1000000; // messages per second
    double bandwidth = throughput * message_size / 1024 / 1024; // MB/s
    
    DEMO_INFO("performance_test: 性能测试结果");
    DEMO_INFO("  消息数量: {}", message_count);
    DEMO_INFO("  消息大小: {} bytes", message_size);
    DEMO_INFO("  总用时: {} μs", duration.count());
    DEMO_INFO("  吞吐量: {} messages/sec", throughput);
    DEMO_INFO("  带宽: {} MB/s", bandwidth);
    DEMO_INFO("  使用msgq模块进行高效存储");
}

int main(int argc, char* argv[]) {
    try {
        if (argc > 1) {
            std::string mode = argv[1];
            
            if (mode == "registry") {
                spdlog_init.set_default_logger("registry");
                run_registry();
            } else if (mode == "publisher") {
                spdlog_init.set_default_logger("publisher");
                run_publisher();
            } else if (mode == "subscriber") {
                spdlog_init.set_default_logger("subscriber");
                if (argc > 2) {
                    std::string name = argv[2];
                    std::vector<std::string> topics;
                    for (int i = 3; i < argc; i++) {
                        topics.push_back(argv[i]);
                    }
                    run_subscriber(name, topics);
                } else {
                    run_subscriber("默认订阅者", {"news"});
                }
            } else if (mode == "perf") {
                spdlog_init.set_default_logger("perf");
                performance_test();
            } else {
                DEMO_INFO("用法: {} [registry|publisher|subscriber|perf|demo]", argv[0]);
                DEMO_INFO("改进版特性:");
                DEMO_INFO("  - 使用msgq模块进行消息存储");
                DEMO_INFO("  - 每个订阅者独立EventFD");
                DEMO_INFO("  - 修复竞争条件问题");
                return 1;
            }
        } else {
            // 默认运行完整演示
            demonstrate_improved_multiprocess_pubsub();
        }
        
    } catch (const std::exception& e) {
        DEMO_CRITICAL("主程序异常: {}", e.what());
        return 1;
    }
    
    return 0;
}
